package jupitermouse.site.example.scala.window

import java.time.ZoneId

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.fs.{SequenceFileWriter, StringWriter}
import org.apache.flink.streaming.connectors.fs.bucketing.{BucketingSink, DateTimeBucketer}

object FileSystemSinkAPP {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val streamSet = env.socketTextStream("localhost",9999)

    val sink = new BucketingSink[String]("file:///E:\\workroom\\learn\\flink\\demo_flink\\doc\\hdfs")

    sink.setBucketer(new DateTimeBucketer("yyyy-MM-dd--HHmm"))
    sink.setWriter(new StringWriter[String]())
//    sink.setBatchSize(1024 * 1024 * 400) // this is 400 MB,
//    sink.setBatchRolloverInterval(20 * 60 * 1000); // this is 20 mins
    sink.setBatchRolloverInterval(20)
    streamSet.addSink(sink)

    env.execute("FileSystemSinkAPP")

  }
}
